package org.dei.perla.aggregates;

import java.util.ArrayList;
import java.util.Iterator;
import org.dei.perla.parser.clauses.LLEveryClause;
import org.dei.perla.parser.clauses.LLEveryEventBased;
import org.dei.perla.parser.clauses.LLEveryTimeBased;
import org.dei.perla.parser.datastructures.Field;
import org.dei.perla.parser.expressions.ConstantCastException;
import org.dei.perla.parser.expressions.ConstantFloat;
import org.dei.perla.parser.util.Duration;
import org.dei.perla.sys.query.llqruntime.datastructures.Record;
import org.dei.perla.utils.waiter.Waitable;
import org.dei.perla.utils.waiter.Waiter;

public class DataAggregator {
	private final Buffer pBufferIn;
	private final Waiter<Buffer> pBufferInWaiter;
	private Thread pBufferListener;
	private AggregationOperator aggOperator;
	private LLEveryClause tLLClause;
	private Field pField;
	
	public DataAggregator(Buffer parBuffer, AggregationOperator operator, LLEveryClause lls, Field pField) {
		System.err.println("Start reading");
		pBufferIn = parBuffer;
		pBufferInWaiter = new Waiter<Buffer>();
		pBufferInWaiter.addWaitable(pBufferIn.getWaitable());
		pBufferInWaiter.start();
		this.aggOperator = operator;
		this.tLLClause = lls;
	    this.pField = pField;
		
	    Duration interval;
	    int samples = 0;
	    String field = pField.getName();
			
        if (tLLClause instanceof LLEveryTimeBased){
            interval = ((LLEveryTimeBased) tLLClause).getInterval();  
            pBufferListener = new Thread(new ReadWindowByTime(interval, field));
        }
        if (tLLClause instanceof LLEveryEventBased){
            samples = ((LLEveryEventBased) tLLClause).getSamples();  
            pBufferListener = new Thread(new ReadWindowBySamples(samples, field));
        }
	   pBufferListener.start();
 	}
 	
	private class ReadWindowBySamples implements Runnable {
		int nSamples;
		String rfield;
		//The frequency in which the DataAggregator is sampling 
		long SamplingFrequency = 2000;
		public ReadWindowBySamples(int wSize, String field) {
			this.nSamples = wSize;
			this.rfield = field;
		}
		public void run() {
			Waitable<Buffer> waitable;
			Buffer buffer;
			do {
				waitable = pBufferInWaiter.waitNext();
				buffer = waitable.getParentComponent();
				// get moving window by samples
				Iterator<Record> it = buffer.getIteratorBackward(buffer.lastRecord().getTimestamp(),
						nSamples);
				if (it == null) {
					try {
						System.err.println("Waiting for enough records "+nSamples);
						//the requiered amount of samples on the buffer
						//For this is needed the DataCollector frequency
						Thread.sleep((long) (SamplingFrequency));
						continue;
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				ArrayList<ConstantFloat> sampleData = new ArrayList<ConstantFloat>();
				double data = 0;
				while (it.hasNext()) {
					Record r = it.next();
					try {
						data = r.getField(rfield).getValueFloat();
					} catch (ConstantCastException e) {
						e.printStackTrace();
					}
					sampleData.add(new ConstantFloat(data));
				}
				ConstantFloat[] sampleArray = new ConstantFloat[sampleData.size()];
				sampleArray = sampleData.toArray(sampleArray);
				ConstantFloat resulti = aggOperator.doOperation(sampleArray);
				try {
					System.err.println("Waiting for ask next window samples");
					Thread.sleep((long) (SamplingFrequency));
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			} while ((true)&&(buffer.lastRecord()!=null));
		}
	}
////// Implementation for the TimeWindow
	
	private class ReadWindowByTime implements Runnable {
		Duration interval;
		String rfield;
		public ReadWindowByTime(Duration wSize, String field) {
			this.interval = wSize;
			this.rfield = field;
		}
		public void run() {
			Waitable<Buffer> waitable;
			Buffer buffer;
			Double dInterval = interval.getValue(); //seconds
			String dTimeUnit = interval.getTimeUnit().toString();
			long tWindow = dInterval.longValue();
			do {
				waitable = pBufferInWaiter.waitNext();
				buffer = waitable.getParentComponent();
				Iterator<Record> it = buffer.getIteratorBackward(buffer.lastRecord().getTimestamp(),
						tWindow);
				if (it == null || it.hasNext()==false) {
					try {
						//Waiting for the Data Collector to complete the requiered samples
						Thread.sleep((long) (tWindow/5));  
						continue;
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				ArrayList<ConstantFloat> sampleData = new ArrayList<ConstantFloat>();
				double data = 0;
				while (it.hasNext()) {
					Record r = it.next();
					try {
						data = r.getField(rfield).getValueFloat();
					} catch (ConstantCastException e) {
						e.printStackTrace();
					}
					sampleData.add(new ConstantFloat(data));
				}
				ConstantFloat[] sampleArray = new ConstantFloat[sampleData.size()];
				sampleArray = sampleData.toArray(sampleArray);
				ConstantFloat resulti = aggOperator.doOperation(sampleArray);
				try {
					System.err.println("Waiting for ask next sample window");
					Thread.sleep((long) (tWindow));
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			} while ((true)&&(buffer.lastRecord()!=null));
		}
	}
}
